package com.hunter.tc.kafka.serialization.avrolocal;

import org.apache.avro.file.DataFileReader;
import org.apache.avro.file.SeekableByteArrayInput;
import org.apache.avro.io.DatumReader;
import org.apache.avro.specific.SpecificDatumReader;
import org.apache.kafka.common.serialization.Deserializer;

import java.io.IOException;
import java.util.Map;

public class AvroLocalDeserializer<T> implements Deserializer<T> {
    private Class<T> tClass;
    //required by the kafka framework
    private DatumReader<T> datumReader;
    private DataFileReader<T> dataFileReader;

    public AvroLocalDeserializer() {

    }

    @Override
    public void configure(Map<String, ?> props, boolean isKey) {
        tClass = (Class<T>) props.get(AvroLocalSerdeConfig.avroPOJOClassName);
        if (tClass == null) {
            throw new IllegalArgumentException("need to set AvroLocalSerdeConfig.avroPOJOClassName");
        }
        datumReader = new SpecificDatumReader<T>();
    }

    @Override
    public T deserialize(String topic, byte[] bytes) {
        try {
            dataFileReader = new DataFileReader<T>(new SeekableByteArrayInput(bytes), datumReader);
            T data = null;
            while (dataFileReader.hasNext()) {
                data = dataFileReader.next(data);
                return data;
            }
        } catch (IOException e) {
            e.printStackTrace();
        }
        return null;
    }

    @Override
    public void close() {
        try {
            dataFileReader.close();
        } catch (IOException e) {
            e.printStackTrace();
        }
    }
}
